package org.apache.ocean.merge;

import java.net.URL;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.commons.lang.StringUtils;
import org.apache.ocean.ClusterUtil;
import org.apache.ocean.OceanXML;
import org.apache.ocean.RefWaitNotNull;
import org.apache.ocean.RestartableTimer;
import org.apache.ocean.TimeoutState;
import org.apache.ocean.name.NameClient;
import org.apache.ocean.name.Node;
import org.apache.ocean.name.NodeUtil;

public class MergeNameClient extends NameClient {
	public static Logger log = Logger.getLogger(MergeNameClient.class.getName());
	private RefWaitNotNull<ClusterModel> clusterRef = new RefWaitNotNull<ClusterModel>();
	private RestartableTimer pollTimer;

	public MergeNameClient(String id, List<URL> nameURLs, EventListener eventListener) throws Exception {
		super(id, nameURLs, eventListener);
		pollTimer = new RestartableTimer("mergeNamePollTimer", new MergeNamePoll(), 5 * 1000, 30 * 1000);
	}
	
	public List<Node> getWriteableNodes(TimeoutState timeoutState) throws Exception {
		ClusterModel clusterModel = clusterRef.get(timeoutState);
		return clusterModel.getWriteableNodes();
	}
	
	public Node getUpdateDocumentNode(String shardID, TimeoutState timeoutState) throws Exception {
		ClusterModel clusterModel = clusterRef.get(timeoutState);
		return clusterModel.getUpdateDocumentNode(shardID);
	}
	
	public static class ClusterModel {
	  private HashMap<String, CellNodes> cells = new HashMap<String, CellNodes>();
		
		public ClusterModel() {
			
		}
		
		public List<Node> getWriteableNodes() throws Exception {
			List<Node> writeableNodes = new ArrayList<Node>();
			for (CellNodes cellNodes : cells.values()) {
				Node writeableNode = cellNodes.getWriteableNode();
				if (writeableNode != null) writeableNodes.add(writeableNode);
			}
			return writeableNodes;
		}
		
		public Node getUpdateDocumentNode(String cellID) throws Exception {
			CellNodes cellNodes = cells.get(cellID);
			if (cellNodes == null) {
        throw new Exception("unknown cell: "+cellID);
			}
			return cellNodes.getWriteableNode();
		}
		
		public Map<String, List<Node>> getReadableNodeMap() {
			Map<String, List<Node>> map = new HashMap<String, List<Node>>();
			for (CellNodes cellNodes : cells.values()) {
				map.put(cellNodes.id, new ArrayList<Node>(cellNodes.getReadableNodes()));
			}
			return map;
		}
		
		public void add(Node node, Date nameServerTimestamp) {
			if (!NodeUtil.verifyNode(node, nameServerTimestamp)) {
				return;
			}
			CellNodes cellNodes = cells.get(node.assignedCellID);
			if (cellNodes == null) {
				cellNodes = new CellNodes(node.assignedCellID, nameServerTimestamp);
			  cells.put(node.assignedCellID, cellNodes);
			}
			cellNodes.add(node);
		}
	}
	
	public static class CellNodes {
		public String id;
		private HashMap<String, Node> readableNodes = new HashMap<String, Node>();
		private Node masterNode;
		private Date serverTimestamp;
    
		public CellNodes(String id, Date serverTimestamp) {
			this.id = id;
			this.serverTimestamp = serverTimestamp;
		}
		
		//public boolean isCellBad() {
		//	getWriteableNode();
		//}
		
		public List<Node> getReadableNodes() {
			return new ArrayList<Node>(readableNodes.values());
		}
		
		public Node getWriteableNode() throws Exception {
			if (masterNode == null) {
				List<Node> nodes = getReadableNodes();
				masterNode = NodeUtil.obtainAgreedMasterNode(nodes, serverTimestamp, log);
			}
			return masterNode;
		}
		
		public void add(Node node) {
			readableNodes.put(node.id, node);
		}
	}
	
	public class MergeNamePoll implements RestartableTimer.TimeoutRunnable {
		public void run(TimeoutState timeoutState) {
			try {
				URL masterNameURL = getMasterNameURL(timeoutState);
				OceanXML response = getNodes(masterNameURL, getDefaultHttpTimeout(timeoutState));
        List<Node> nodes = response.bodyToList(new Node.Transform());
        //ClusterModel clusterModel = clusterRef.get();
        ClusterModel clusterModel = new ClusterModel();
				for (Node node : nodes) {
					//currentCluster.add(node, response.timestamp);
					clusterModel.add(node, response.timestamp);
        }
				clusterRef.set(clusterModel);
			} catch (Throwable throwable) {
        log.log(Level.SEVERE, "", throwable);
			}
		}
	}

	public Map<String, List<Node>> getReadableCellNodes(TimeoutState timeoutState) throws Exception {
		ClusterModel clusterModel = clusterRef.get(timeoutState);
		return clusterModel.getReadableNodeMap();
	}
}
